package com.spx.sql;

import com.spx.chapter05.pojo.Event;
import com.spx.util.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * create by undeRdoG on  2022-05-10  14:05
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class SimpleTableExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );


        // 1.创建流式的表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2.将一个流转换成表(POJO中的属性，就是表的字段)
        Table eventTable = tableEnv.fromDataStream(eventStream);

        // 直接写SQL
        Table resultTable = tableEnv.sqlQuery("select user,url from " + eventTable);

        // 基于Table的转换
        Table resultTable2 = eventTable.select($("user"), $("url"))
                .where($("user").isEqual("Alice"));

        // 将表转换回流打印
        tableEnv.toDataStream(resultTable).print("result");

        tableEnv.toDataStream(resultTable2).print("result2");

        env.execute();

    }
}
